package org.example.seckilljava.service;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.example.seckilljava.config.KafkaConfig;
import org.example.seckilljava.dto.SeckillMessage;
import org.example.seckilljava.entity.Order;
import org.example.seckilljava.entity.Quota;
import org.example.seckilljava.entity.SeckillRecord;
import org.example.seckilljava.entity.UserQuota;
import org.example.seckilljava.mapper.*;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Service
public class KafkaConsumerService {

    @Resource
    private ObjectMapper objectMapper;
    @Resource
    private OrderMapper orderMapper;
    @Resource
    private SeckillRecordMapper seckillRecordMapper;
    @Resource
    private UserQuotaMapper userQuotaMapper;
    @Resource
    private QuotaMapper quotaMapper;
    @Resource
    private SeckillStockMapper seckillStockMapper;

    @Transactional
    @KafkaListener(
            topics = KafkaConfig.SECKILL_TOPIC,
            groupId = "${spring.kafka.consumer.group-id}",
            containerFactory = "kafkaListenerContainerFactory"
    )
    public void consumeSeckillMessage_v2_old(List<String> messages, Acknowledgment ack) {
        try {
            log.info("收到一批消息，数量: {}", messages.size());

            for (String message : messages) {
                try {
                    SeckillMessage seckillMessage = objectMapper.readValue(message, SeckillMessage.class);
                    Order order = seckillMessage.getOrder();
                    SeckillRecord seckillRecord = seckillMessage.getSeckillRecord();
                    log.info("处理秒杀消息: {}", message);
                    // 1 判断限额
                    Boolean quotaFlag = judgeQuota_v1(order.getBuyer(), order.getGoodsId(), order.getNum());
                    if (!quotaFlag) { // 回滚
                        log.info("限额不满足，可能是超卖或并发冲突");
                        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                        return ;
                    }
                    // 2 操作数据库
                    //  2-1 用户限额更新
                    Integer affectedRows = userQuotaMapper.incrKilledNum(order.getBuyer(), order.getGoodsId(), order.getNum());
                    if (affectedRows == 0) {
                        log.info("用户限额更新失败，可能是超卖或并发冲突");
                        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                        return;
                    }
                    //  2-2 全局限额更新
                    affectedRows = quotaMapper.incrKilledNum(order.getGoodsId(), order.getNum());
                    if (affectedRows == 0) {
                        log.info("全局限额更新失败，可能是超卖或并发冲突");
                        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                        return;
                    }
                    //  2-3 库存更新
                    affectedRows = seckillStockMapper.descStock(order.getGoodsId(), order.getNum());
                    if (affectedRows == 0) {
                        log.info("库存更新失败，可能是超卖或并发冲突");
                        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                        return;
                    }
                    // 3 插入订单和记录
                    orderMapper.save(order);
                    seckillRecordMapper.save(seckillRecord);
                } catch (Exception e) {
                    log.error("单条秒杀消息处理失败，回滚事务: {}", message, e);
                    throw e;  // 抛出异常触发整批事务回滚
                }
            }
            // 只有成功了才提交 Kafka 偏移量
            ack.acknowledge();
        } catch (Exception e) {
            log.error("批量处理消息失败，事务将回滚，Kafka消息不会确认", e);
            // 不手动 ack，会触发 Kafka 自动重试或进入死信队列
        }
    }

    /**
     * 新的消费消息的方法：少卖情况得到很大缓解
     * @param messages
     * @param ack
     */
//    @Transactional
//    @KafkaListener(
//            topics = KafkaConfig.SECKILL_TOPIC,
//            groupId = "${spring.kafka.consumer.group-id}",
//            containerFactory = "kafkaListenerContainerFactory"
//    )
    public void consumeSeckillMessage_v2(List<String> messages, Acknowledgment ack) {
        try {
            log.info("收到一批消息，数量: {}", messages.size());
            // 用Map<Long, Integer>来记录这些消息里的<goodsId, totalNum>
            Map<Long, Integer> map = new HashMap<>();
            for (String message : messages) {
                try {
                    SeckillMessage seckillMessage = objectMapper.readValue(message, SeckillMessage.class);
                    Order order = seckillMessage.getOrder();
                    SeckillRecord seckillRecord = seckillMessage.getSeckillRecord();
                    Long goodsId = order.getGoodsId();
                    Integer num = order.getNum();
                    // 1 判断限额: 全局限额不满足，直接回滚整批消息即可
                    Boolean quotaFlag = judgeQuota_v1(order.getBuyer(), goodsId, num);
                    if (!quotaFlag) { // 回滚
                        log.info("限额不满足，可能是超卖或并发冲突");
                        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                        return;
                    }
                    // 2 操作数据库
                    //  2-1 用户限额更新: 用户限额不满足，continue
                    Integer affectedRows = userQuotaMapper.incrKilledNum(order.getBuyer(), goodsId, num);
                    if (affectedRows == 0) {
                        log.warn("用户限额更新失败，可能是超卖或并发冲突");
                        continue;
                    }
                    //  2-2 全局限额&库存的变更记录
                    map.put(goodsId, map.getOrDefault(goodsId, 0) + num);
                    // 3 插入订单和记录
                    orderMapper.save(order);
                    seckillRecordMapper.save(seckillRecord);
                } catch (Exception e) {
                    log.error("单条秒杀消息处理失败，回滚事务: {}", message, e);
                    throw e;  // 抛出异常触发整批事务回滚
                }
            }
            // 更新全局限额、库存
            for (Long key : map.keySet()) {
                int value = map.get(key);
                Integer affectedRows = quotaMapper.incrKilledNum(key, value);
                if (affectedRows == 0) {
                    log.info("全局限额更新失败，可能是超卖或并发冲突，商品id={}，购买总数={}", key, value);
                    TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                    return;
                }
                affectedRows = seckillStockMapper.descStock(key, value);
                if (affectedRows == 0) {
                    log.info("库存更新失败，可能是超卖或并发冲突，商品id={}，购买总数={}", key, value);
                    TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                    return;
                }
            }
            // 只有成功了才提交 Kafka 偏移量
            ack.acknowledge();
        } catch (Exception e) {
            log.error("批量处理消息失败，事务将回滚，Kafka消息不会确认", e);
            // 不手动 ack，会触发 Kafka 自动重试或进入死信队列
        }
    }


    /**
     * 和 v1 同样的判断方法
     * @param userID
     * @param goodsID
     * @param num
     * @return
     */
    private Boolean judgeQuota_v1(long userID, long goodsID, Integer num) {
        Integer userKilledNum; // 用户已经秒杀的商品数
        Integer userQuotaNum; // 用户秒杀限额数
        Integer globalKilledNum; // 全局已经秒杀的商品数
        Integer globalQuotaNum; // 全局商品限额数
        int userFQ = -1, globalFQ = -1;
        // 1. 拿到用户限额数据和商品限额数据
        UserQuota userQuota = userQuotaMapper.getUserGoodsUserQuota(userID, goodsID);
        Quota globalQuota = quotaMapper.getGoodsQuota(goodsID);
        // 2. 如果有数据，复制
        if (userQuota == null) {
            userQuota = new UserQuota();
            userQuota.setGoodsId(goodsID);
            userQuota.setKilledNum(0);
            userQuota.setNum(1); // 默认限额一个
            userQuota.setUserId(userID);
            userQuotaMapper.save(userQuota);
        }
        if (globalQuota == null) {
            globalQuota = new Quota();
            globalQuota.setGoodsId(goodsID);
            globalQuota.setNum(-1); // 默认不限额
            globalQuota.setKilledNum(0);
            quotaMapper.save(globalQuota);
        }
        userKilledNum = userQuota.getKilledNum();
        userQuotaNum = userQuota.getNum();
        globalKilledNum = globalQuota.getKilledNum();
        globalQuotaNum = globalQuota.getNum();
        // 3. 确定限额情况
        if (userQuotaNum != -1) {
            userFQ = userQuotaNum - userKilledNum;
        }
        if (globalQuotaNum != -1) {
            globalFQ = globalQuotaNum - globalKilledNum;
        }
        // 4. 所有情况
        if (userFQ == -1 && globalFQ == -1)
            return true;
        if (globalFQ == -1) // 如果 global 不限额
            return userFQ >= num;
        if (userFQ == -1) // 如果 user 不限额
            return globalFQ >= num;
        return userFQ >= num && globalFQ >= num;
    }
}
